package com.raylu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.raylu.realtime.app.template.DimAsyncFunction;
import com.raylu.realtime.bean.*;
import com.raylu.realtime.utils.ClickHouseUtil;
import com.raylu.realtime.utils.DateUtil;
import com.raylu.realtime.utils.KafkaSourceUtil;
import com.raylu.realtime.utils.PropertiesUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Collections;
import java.util.HashSet;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

/**
 * Description:
 * <p>
 * Create by lucienoz on 2022/1/7.
 * Copyright © 2022 lucienoz. All rights reserved.
 */
public class ProductStatsApp {
    public static void main(String[] args) throws Exception {
        Properties load = PropertiesUtil.load("config.properties");
        //TODO 1. 准备运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        //TODO 2. 配置检查点
//        env.enableCheckpointing(5000);
//        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//        env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L);
//        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
//        env.setStateBackend(new FsStateBackend(load.getProperty("product.stats.app.fsstatebackend.url")));
//        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, org.apache.flink.api.common.time.Time.seconds(10L), org.apache.flink.api.common.time.Time.hours(10L)));
//        System.setProperty("HADOOP_USER_NAME", "raylu");
        //TODO 3. 订阅Kafka主题，并将各个主题的数据转换成ProductStats对象
        //TODO 3.1 点击 多维分析 page_log直接可求 dwd
        SingleOutputStreamOperator<ProductStats> clickDS = env
                .addSource(KafkaSourceUtil.getKafkaSource(load.getProperty("product.stats.app.kafka.source-topic1"), load.getProperty("product.stats.app.kafka.group-id")))
                .process(new ProcessFunction<String, ProductStats>() {
                    @Override
                    public void processElement(String value, ProcessFunction<String, ProductStats>.Context ctx, Collector<ProductStats> out) throws Exception {
                        PageBean pageBean = JSON.parseObject(value, PageBean.class);
                        if ("good_detail".equals(pageBean.page_id)) { //如果Page ID 为good_detail，则认为此页面为点击进入页面
                            ProductStats build = ProductStats.builder()
                                    .sku_id(pageBean.item)
                                    .click_cnt(1L)
                                    .ts(pageBean.ts)
                                    .build();
                            out.collect(build);
                        }
                    }
                });
        //TODO 3.2 曝光 多维分析 DWD_DISPLAY_LOG dwd
        SingleOutputStreamOperator<ProductStats> displayDS = env
                .addSource(KafkaSourceUtil.getKafkaSource(load.getProperty("product.stats.app.kafka.source-topic2"), load.getProperty("product.stats.app.kafka.group-id")))
                .process(new ProcessFunction<String, ProductStats>() {
                    @Override
                    public void processElement(String value, ProcessFunction<String, ProductStats>.Context ctx, Collector<ProductStats> out) throws Exception {
                        DisplayBean displayBean = JSON.parseObject(value, DisplayBean.class);
                        if ("sku_id".equals(displayBean.item_type)) {
                            ProductStats build = ProductStats.builder()
                                    .sku_id(displayBean.item)
                                    .display_cnt(1L)
                                    .ts(displayBean.getTs())
                                    .build();
                            out.collect(build);
                        }
                    }
                });

        //TODO 3.3 收藏 多维分析 DWD_FAVOR_INFO           dwd
        SingleOutputStreamOperator<ProductStats> favorDS = env
                .addSource(KafkaSourceUtil.getKafkaSource(load.getProperty("product.stats.app.kafka.source-topic3"), load.getProperty("product.stats.app.kafka.group-id")))
                .process(new ProcessFunction<String, ProductStats>() {
                    @Override
                    public void processElement(String value, ProcessFunction<String, ProductStats>.Context ctx, Collector<ProductStats> out) throws Exception {
                        JSONObject jsonObject = JSON.parseObject(value);
                        if (!"1".equals(jsonObject.getString("is_cancel"))) {
                            ProductStats build = ProductStats.builder()
                                    .sku_id(jsonObject.getString("sku_id"))
                                    .favor_cnt(1L)
                                    .ts(DateUtil.getEpochMilli(jsonObject.getString("create_time")))
                                    .build();
                            out.collect(build);
                        }
                    }
                });
        //TODO 3.4 加入购物车  多维分析  DWD_CART_INFO  dwd
        SingleOutputStreamOperator<ProductStats> cartDS = env
                .addSource(KafkaSourceUtil.getKafkaSource(load.getProperty("product.stats.app.kafka.source-topic4"), load.getProperty("product.stats.app.kafka.group-id")))
                .process(new ProcessFunction<String, ProductStats>() {
                    @Override
                    public void processElement(String value, ProcessFunction<String, ProductStats>.Context ctx, Collector<ProductStats> out) throws Exception {
                        JSONObject jsonObject = JSON.parseObject(value);
                        ProductStats build = ProductStats.builder()
                                .sku_id(jsonObject.getString("sku_id"))
                                .add_cart_cnt(1L)
                                .ts(DateUtil.getEpochMilli(jsonObject.getString("create_time")))
                                .build();
                        out.collect(build);

                    }
                });
        //TODO 3.5 下单 可视化大屏     DWM_ORDER_WIDE  dwm
        SingleOutputStreamOperator<ProductStats> orderDS = env
                .addSource(KafkaSourceUtil.getKafkaSource(load.getProperty("product.stats.app.kafka.source-topic5"), load.getProperty("product.stats.app.kafka.group-id")))
                .process(new ProcessFunction<String, ProductStats>() {
                    @Override
                    public void processElement(String value, ProcessFunction<String, ProductStats>.Context ctx, Collector<ProductStats> out) throws Exception {
                        OrderWide orderWide = JSON.parseObject(value, OrderWide.class);
                        ProductStats build = ProductStats.builder()
                                .sku_id(orderWide.getSku_id())
                                .orderSet(new HashSet(Collections.singleton(orderWide.getSku_id())))
                                .order_amount(orderWide.getSplit_total_amount())
                                .ts(orderWide.getTs())
                                .build();
                        out.collect(build);
                    }
                });
//        orderDS.print("DWM_ORDER_WIDE>>>>");
        //TODO 3.6 支付 多维分析 DWM_PAYMENT_WIDE dwm
        SingleOutputStreamOperator<ProductStats> paidDS = env
                .addSource(KafkaSourceUtil.getKafkaSource(load.getProperty("product.stats.app.kafka.source-topic6"), load.getProperty("product.stats.app.kafka.group-id")))
                .process(new ProcessFunction<String, ProductStats>() {
                    @Override
                    public void processElement(String value, ProcessFunction<String, ProductStats>.Context ctx, Collector<ProductStats> out) throws Exception {
                        PaymentWide paymentWide = JSON.parseObject(value, PaymentWide.class);
                        ProductStats build = ProductStats.builder()
                                .sku_id(paymentWide.getSku_id())
                                .paidSet(new HashSet(Collections.singleton(paymentWide.getSku_id())))
                                .paid_amount(paymentWide.getSplit_total_amount())
                                .ts(paymentWide.getTs())
                                .build();
                        out.collect(build);
                    }
                });
//        paidDS.print("DWM_PAYMENT_WIDE>>>>");
        //TODO 3.7 退款 多维分析 DWD_REFUND_PAYMENT dwd
        SingleOutputStreamOperator<ProductStats> refundDS = env
                .addSource(KafkaSourceUtil.getKafkaSource(load.getProperty("product.stats.app.kafka.source-topic7"), load.getProperty("product.stats.app.kafka.group-id")))
                .process(new ProcessFunction<String, ProductStats>() {
                    @Override
                    public void processElement(String value, ProcessFunction<String, ProductStats>.Context ctx, Collector<ProductStats> out) throws Exception {
                        JSONObject jsonObject = JSON.parseObject(value);
                        ProductStats build = ProductStats.builder()
                                .sku_id(jsonObject.getString("sku_id"))
                                .refundSet(new HashSet(Collections.singleton(jsonObject.getString("sku_id"))))
                                .refund_amount(jsonObject.getBigDecimal("total_amount"))
                                .ts(DateUtil.getEpochMilli(jsonObject.getString("create_time")))
                                .build();
                        out.collect(build);
                    }
                });
        //TODO 3.8 评价 多维分 DWD_COMMENT_INFO dwd
        SingleOutputStreamOperator<ProductStats> commentDS = env
                .addSource(KafkaSourceUtil.getKafkaSource(load.getProperty("product.stats.app.kafka.source-topic8"), load.getProperty("product.stats.app.kafka.group-id")))
                .process(new ProcessFunction<String, ProductStats>() {
                    @Override
                    public void processElement(String value, ProcessFunction<String, ProductStats>.Context ctx, Collector<ProductStats> out) throws Exception {
                        JSONObject jsonObject = JSON.parseObject(value);
                        ProductStats build = ProductStats.builder()
                                .sku_id(jsonObject.getString("sku_id"))
                                .comment_cnt(1L)
                                .good_comment_cnt("1201".equals(jsonObject.getString("appraise")) ? 1L : 0L)
                                .ts(DateUtil.getEpochMilli(jsonObject.getString("create_time")))
                                .build();
                        out.collect(build);
                    }
                });


        //TODO 4. 将各个主题的数据union成一条流,并设置其水位线
        SingleOutputStreamOperator<ProductStats> productStatsSingleDS = clickDS
                .union(
                        displayDS,
                        favorDS,
                        cartDS,
                        orderDS,
                        paidDS,
                        refundDS,
                        commentDS
                )
                .assignTimestampsAndWatermarks(WatermarkStrategy.<ProductStats>forBoundedOutOfOrderness(Duration.ofSeconds(3L))
                        .withTimestampAssigner(new SerializableTimestampAssigner<ProductStats>() {
                            @Override
                            public long extractTimestamp(ProductStats element, long recordTimestamp) {
                                return element.getTs();
                            }
                        }).withIdleness(Duration.ofMinutes(1L))
                );


        //TODO 5. 将各个流按照最细的粒度sku_id进行分组，然后进行滚动开窗，将各个窗口的数据进行指标聚合
        SingleOutputStreamOperator<ProductStats> reduceDS = productStatsSingleDS.keyBy(r -> r.getSku_id())
                .window(TumblingEventTimeWindows.of(Time.seconds(10L)))
                .reduce(new ReduceFunction<ProductStats>() {
                    @Override
                    public ProductStats reduce(ProductStats value1, ProductStats value2) throws Exception {
                        value1.setClick_cnt(value1.getClick_cnt() + value2.getClick_cnt());
                        value1.setComment_cnt(value1.getComment_cnt() + value2.getComment_cnt());
                        value1.setDisplay_cnt(value1.getDisplay_cnt() + value2.getDisplay_cnt());
                        value1.setFavor_cnt(value1.getFavor_cnt() + value2.getFavor_cnt());
                        value1.setAdd_cart_cnt(value1.getAdd_cart_cnt() + value2.getAdd_cart_cnt());
                        value1.setOrder_amount(value1.getOrder_amount().add(value2.getOrder_amount()));
                        value1.setGood_comment_cnt(value1.getGood_comment_cnt() + value2.getGood_comment_cnt());
                        value1.setPaid_amount(value1.getPaid_amount().add(value2.getPaid_amount()));
                        value1.setRefund_amount(value1.getRefund_amount().add(value2.getRefund_amount()));

                        value1.getOrderSet().addAll(value2.getOrderSet());
                        value1.getPaidSet().addAll(value2.getPaidSet());
                        value1.getRefundSet().addAll(value2.getRefundSet());

                        value1.setOrder_cnt((long) value1.getOrderSet().size());
                        value1.setPaid_cnt((long) value1.getPaidSet().size());
                        value1.setRefund_cnt((long) value1.getRefundSet().size());

                        System.out.println("value1 = " + value1);
                        return value1;
                    }
                }, new ProcessWindowFunction<ProductStats, ProductStats, String, TimeWindow>() {
                    @Override
                    public void process(String s, ProcessWindowFunction<ProductStats, ProductStats, String, TimeWindow>.Context context, Iterable<ProductStats> elements, Collector<ProductStats> out) throws Exception {
                        long start = context.window().getStart();
                        long end = context.window().getEnd();
                        String formatDateTime = DateUtil.getFormatDateTime(context.currentWatermark());
                        for (ProductStats element : elements) {
                            element.setStt(DateUtil.getFormatDateTime(start));
                            element.setEdt(DateUtil.getFormatDateTime(end));
                            element.setTs(System.currentTimeMillis());
                            System.out.println("element.getOrder_amount() = " + element.getOrder_amount() + " getPaid_amount() = " + element.getPaid_amount() + " getRefund_amount() = " + element.getRefund_amount());
                            System.out.println("element.getOrder_cnt() = " + element.getOrder_cnt() + " getPaid_cnt() = " + element.getPaid_cnt() + " getRefund_cnt() = " + element.getRefund_cnt());
                            System.out.println(formatDateTime);

                            out.collect(element);
                        }

                    }
                });

        reduceDS.print("reduceDS>>>>");

        //TODO 6. 将其他维度数据进行关联取值
        //TODO 6.1 //sku info
        SingleOutputStreamOperator<ProductStats> skuDS = AsyncDataStream.unorderedWait(reduceDS,
                new DimAsyncFunction<ProductStats>() {
                    @Override
                    public Tuple2<String, String>[] getConditions() {
                        Tuple2[] tuple2s = {Tuple2.of("sku_id", "id")};
                        return tuple2s;
                    }

                    @Override
                    public String getTable() {
                        return load.getProperty("base.db.app.cdc.phoenix.schem") + "." + "DIM_SKU_INFO";
                    }

                    @Override
                    public void join(ProductStats input, JSONObject jsonObject) {
                        input.setSku_name(jsonObject.getString("SKU_NAME"));
                        input.setSpu_id(jsonObject.getString("SPU_ID"));
                    }
                }, 60, TimeUnit.SECONDS);

        //TODO 6.2 //spu info
        SingleOutputStreamOperator<ProductStats> spuDS = AsyncDataStream.unorderedWait(skuDS,
                new DimAsyncFunction<ProductStats>() {
                    @Override
                    public Tuple2<String, String>[] getConditions() {
                        Tuple2[] tuple2s = {Tuple2.of("spu_id", "id")};
                        return tuple2s;
                    }

                    @Override
                    public String getTable() {
                        return load.getProperty("base.db.app.cdc.phoenix.schem") + "." + "DIM_SPU_INFO";
                    }

                    @Override
                    public void join(ProductStats input, JSONObject jsonObject) {
                        input.setSpu_name(jsonObject.getString("SPU_NAME"));
                        input.setCategory3_id(jsonObject.getString("CATEGORY3_ID"));
                        input.setTm_id(jsonObject.getString("TM_ID"));
                    }
                }, 60, TimeUnit.SECONDS);

        //TODO 6.2 //TM info
        SingleOutputStreamOperator<ProductStats> tmDS = AsyncDataStream.unorderedWait(spuDS,
                new DimAsyncFunction<ProductStats>() {
                    @Override
                    public Tuple2<String, String>[] getConditions() {
                        Tuple2[] tuple2s = {Tuple2.of("tm_id", "id")};
                        return tuple2s;
                    }

                    @Override
                    public String getTable() {
                        return load.getProperty("base.db.app.cdc.phoenix.schem") + "." + "DIM_BASE_TRADEMARK";
                    }

                    @Override
                    public void join(ProductStats input, JSONObject jsonObject) {
                        input.setTm_name(jsonObject.getString("TM_NAME"));
                    }
                }, 60, TimeUnit.SECONDS);
        //TODO 6.3 //C3 info
        SingleOutputStreamOperator<ProductStats> c3DS = AsyncDataStream.unorderedWait(tmDS,
                new DimAsyncFunction<ProductStats>() {
                    @Override
                    public Tuple2<String, String>[] getConditions() {
                        Tuple2[] tuple2s = {Tuple2.of("category3_id", "id")};
                        return tuple2s;
                    }

                    @Override
                    public String getTable() {
                        return load.getProperty("base.db.app.cdc.phoenix.schem") + "." + "DIM_BASE_CATEGORY3";
                    }

                    @Override
                    public void join(ProductStats input, JSONObject jsonObject) {
                        input.setCategory3_name(jsonObject.getString("NAME"));
                        input.setCategory2_id(jsonObject.getString("CATEGORY2_ID"));

                    }
                }, 60, TimeUnit.SECONDS);

        //TODO 6.4 //C2 info
        SingleOutputStreamOperator<ProductStats> c2DS = AsyncDataStream.unorderedWait(c3DS,
                new DimAsyncFunction<ProductStats>() {
                    @Override
                    public Tuple2<String, String>[] getConditions() {
                        Tuple2[] tuple2s = {Tuple2.of("category2_id", "id")};
                        return tuple2s;
                    }

                    @Override
                    public String getTable() {
                        return load.getProperty("base.db.app.cdc.phoenix.schem") + "." + "DIM_BASE_CATEGORY2";
                    }

                    @Override
                    public void join(ProductStats input, JSONObject jsonObject) {
                        input.setCategory2_name(jsonObject.getString("NAME"));
                        input.setCategory1_id(jsonObject.getString("CATEGORY1_ID"));
                    }
                }, 60, TimeUnit.SECONDS);

        //TODO 6.5 //C1 info
        SingleOutputStreamOperator<ProductStats> c1DS = AsyncDataStream.unorderedWait(c2DS,
                new DimAsyncFunction<ProductStats>() {
                    @Override
                    public Tuple2<String, String>[] getConditions() {
                        Tuple2[] tuple2s = {Tuple2.of("category1_id", "id")};
                        return tuple2s;
                    }

                    @Override
                    public String getTable() {
                        return load.getProperty("base.db.app.cdc.phoenix.schem") + "." + "DIM_BASE_CATEGORY1";
                    }

                    @Override
                    public void join(ProductStats input, JSONObject jsonObject) {
                        input.setCategory1_name(jsonObject.getString("NAME"));
                    }
                }, 60, TimeUnit.SECONDS);

        //TODO 7. 将最后的数据sink到ClickHouse数据库
        c1DS.print("result>>>");
        c1DS.addSink(ClickHouseUtil.getJdbcSink("INSERT INTO gmall.dws_product_stats (stt, edt, sku_id, sku_name, spu_id, tm_id, spu_name, category3_id, tm_name, category3_name, category2_id, category2_name, category1_id, category1_name, click_cnt, display_cnt, favor_cnt, add_cart_cnt, order_cnt, order_amount, paid_cnt, paid_amount, refund_cnt, refund_amount, comment_cnt, good_comment_cnt, ts) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"));


        env.execute();


    }
}
